package com.taurus.permanent.io;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.DatagramChannel;
import java.util.concurrent.Executor;

import com.taurus.core.entity.ITObject;
import com.taurus.core.entity.TObject;
import com.taurus.core.util.Logger;
import com.taurus.core.util.Utils;
import com.taurus.permanent.TPServer;
import com.taurus.permanent.core.BitSwarmEngine;
import com.taurus.permanent.core.SessionManager;
import com.taurus.permanent.data.Packet;
import com.taurus.permanent.data.Session;
import com.taurus.permanent.data.TransportType;
import com.taurus.permanent.util.CodecException;

/**
 * 读写网络字节流
 * 
 */
public class IOHandler {
	private final BinaryIoHandler binHandler;
	private final Logger          logger;
	private final Executor        systemThreadPool;

	public IOHandler() {
		logger = Logger.getLogger(getClass());
		binHandler = new BinaryIoHandler(this);

		TPServer taurus = TPServer.me();
		this.systemThreadPool = taurus.getSystemExecutor();
	}

	public void onDataRead(DatagramChannel channel, SocketAddress address, byte[] data) {
		this.systemThreadPool.execute(new UdpIOExecutor(channel, address, data));
	}

	/**
	 * 读取网络包字节流
	 * 
	 * @param session
	 * @param data
	 */
	public void onDataRead(Session session, byte[] data) {
		if ((data == null) || (data.length < 1)) {
			throw new IllegalArgumentException("Unexpected null or empty byte array!");
		}

		PacketReadState readState = (PacketReadState) session.getSystemProperty(Session.PACKET_READ_STATE);
		if (readState == null) {
			session.setSystemProperty(Session.PACKET_READ_STATE, PacketReadState.WAIT_NEW_PACKET);
		}

		binHandler.handleRead(session, data);
	}

	/**
	 * 数据写入网络字节流，发送给指定客户端
	 * 
	 * @param packet
	 */
	public void onDataWrite(Packet packet) {
		if (packet.getRecipients().size() > 0) {
			try {
				this.binHandler.handleWrite(packet);
			} catch (Exception e) {
				logger.error(e);
			}
		}
	}

	/**
	 * 统计丢包数量
	 * 
	 * @return
	 */
	public long getReadPackets() {
		return binHandler.getReadPackets();
	}

	/**
	 * 统计丢包数量
	 * 
	 * @return
	 */
	public long getIncomingDroppedPackets() {
		return binHandler.getIncomingDroppedPackets();
	}

	private Session validateUDPRequest(String senderIP, int senderPort, Session session) throws CodecException {
//	    if (!packet.containsKey("u"))
//	      throw new CodecException("Missing userId"); 
//	    if (!packet.containsKey("i"))
//	      throw new CodecException("Missing packet id"); 
//	    int sessionId = packet.getInt(SystemController.CONNECT_TOKE);
//	    User sender = this.sfs.getUserManager().getUserById(userId);
//	    if (sender == null)
//	      throw new CodecException("User does not exist, id: " + userId); 
//	    Session session =SessionManager.getInstance().getSessionByValidate(sessionId);
		if (!session.getAddress().equals(senderIP))
			throw new CodecException(String.format("Sender IP doesn't match TCP session address: " + "%s != %s",
					senderIP, session.getAddress()));
		Integer sessionUdpPort = (Integer) session.getSystemProperty("UDPPort");
		if (sessionUdpPort == null) {
			session.setSystemProperty("UDPPort", Integer.valueOf(senderPort));
		} else if (senderPort != sessionUdpPort.intValue()) {
			logger.error("senderPort::::::::::" + senderPort);
			session.setSystemProperty("UDPPort", Integer.valueOf(senderPort));
		}
		return session;
	}

	private final class UdpIOExecutor implements Runnable {
		DatagramChannel channel;

		SocketAddress   address;

		byte[]          data;

		public UdpIOExecutor(DatagramChannel channel, SocketAddress address, byte[] data) {
			this.channel = channel;
			this.address = address;
			this.data = data;
		}

		public void run() {
			String senderIP = null;
			int senderPort = 0;
			try {
				if (this.data.length < 15)
					throw new RuntimeException("Packet too small: " + this.data.length + " bytes");

				int dataSize = 0;
				int sessionId = 0;
				boolean compressed = data[0] > 0;
				String[] adrData = this.address.toString().split("\\:");
				senderIP = adrData[0].substring(1);
				senderPort = Integer.parseInt(adrData[1]);
				for (int i = 1; i < 5; i++) {
					int pow256 = (int) Math.pow(256, 4 - i);
					int intByte = data[i] & 0xFF;
					dataSize += pow256 * intByte;
				}

				for (int i = 5; i < 9; i++) {
					int pow256 = (int) Math.pow(256, 8 - i);
					int intByte = data[i] & 0xFF;
					sessionId += pow256 * intByte;
				}
				Session session = SessionManager.me().getSessionById(sessionId);
				if (session == null || !session.isValidate()) {
					throw new CodecException("Session does not exist, hashcode: " + sessionId);
				}
				int maxPacketSize = TPServer.me().getConfig().maxPacketSize;
				if (dataSize > maxPacketSize) {
					throw new IllegalArgumentException(
							String.format("Incoming request size too large: %s, Current limit: %s, From: %s",
									dataSize, maxPacketSize, session));
				}

				byte[] userData = new byte[this.data.length - 9];
				System.arraycopy(this.data, 9, userData, 0, userData.length);
				if (userData.length != dataSize)
					throw new RuntimeException(
							"Packet truncated. Expected: " + dataSize + ", only got: " + userData.length);
				if (compressed) {
					try {
						userData = Utils.uncompress(userData);
					} catch (IOException e) {
						throw new RuntimeException(e);
					}
				}

				validateUDPRequest(senderIP, senderPort, session);
				session.setDatagramChannel(channel);
				session.addReadBytes(this.data.length);

				ITObject obj = TObject.newFromBinaryData(userData);
				Packet packet = new Packet();
				packet.setData(obj);
				packet.setSender(session);
				packet.setTransportType(TransportType.UDP);
				BitSwarmEngine.getInstance().getProtocolHandler().onPacketRead(packet);
			} catch (RuntimeException err) {
				logger.warn("Problems decoding UDP packet from: " + senderIP + ":" + senderPort + ", " + err);
			} catch (CodecException codecErr) {
				logger.warn(String.format("Discard UDP packet from %s:%s, reason: %s ", senderIP,
						senderPort, codecErr.getMessage()));
			}
		}
	}
}
